package com.lvmama.rhino.common.utils.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by yuanxiaofeng on 2017/4/18.
  */
trait StreamingApplication {
  var appName: String

  var sparkConfig: Map[String, String]

  def streamingBatchDuration: Long

  def streamingCheckpointDir: String

  def withSparkStreamingContext(f: (SparkContext, StreamingContext) => Unit): StreamingContext = {
    val conf = new SparkConf()
    sparkConfig.foreach{case (k, v) => conf.setIfMissing(k, v)}

    val sc = new SparkContext(conf)
    val ssc = new StreamingContext(sc, Seconds(streamingBatchDuration))
    ssc.checkpoint(streamingCheckpointDir)
    f(sc, ssc)
    ssc
  }
}
